package com.jwater.core.manager.job;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.conf.JobConfParse;
import com.jwater.core.manager.common.model.JobInfo;
import com.jwater.core.manager.job.timer.CronTimer;
import com.jwater.core.manager.job.timer.TimerTask;
import com.jwater.core.manager.master.MasterNodeImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @description:
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class JobScheduler implements TimerTask {
    protected static Logger logger = LoggerFactory.getLogger(JobScheduler.class);
    private JobInfo jobInfo;
    private JobConfParse jobConf;
    private MasterNodeImpl masterNode;
    private CronTimer timer;
    private volatile boolean run = false;

    public JobScheduler(JobInfo jobInfo, JobConfParse jobConf, MasterNodeImpl masterNode) {
        this.jobInfo = jobInfo;
        this.jobConf = jobConf;
        this.masterNode = masterNode;
        timer = new CronTimer(this, jobInfo.getCron());
    }

    /**
     * 启动定时器
     */
    private synchronized void startTimer() {
        if (!run) {
            run = true;
            timer.start();
            logger.info("start JobScheduler for:" + this.jobInfo);
        }
    }

    /**
     * 启动定时器
     */
    public void startScheduler() {
        startTimer();
    }

    /**
     * 停止定时器
     */
    public void stopScheduler() {
        if (run) {
            run = false;
            timer.stop();
            logger.info("stop JobScheduler for:" + this.jobInfo);
        }
    }

    /**
     * 尝试停止调度
     *
     * @return
     */
    public boolean tryStopScheduler() {
        List<JobInfo> jobInfos = jobConf.getGroupJobInfo(jobInfo.getRegionName(), jobInfo.getDeployName(),
                jobInfo.getGroupName());
        for (JobInfo job : jobInfos) {
            String jobKey = DataKeys.getJobKey(job.getRegionName(), job.getDeployName(), job.getGroupName(),
                    job.getJobName());
            String run = masterNode.getDataItem(DataKeys.KEY_JOB_RUN, jobKey);
            if (run != null) {
                return false;
            }
        }
        this.stopScheduler();
        return true;
    }

    /**
     * 运行下一个job
     *
     * @param jobInfo
     * @param time
     */
    public void runNextJob(JobInfo jobInfo, long time) {
        List<JobInfo> jobInfos = jobConf.getGroupJobInfo(jobInfo.getRegionName(), jobInfo.getDeployName(),
                jobInfo.getGroupName());
        if (jobInfos == null || jobInfos.isEmpty()) {
            return;
        }
        Collections.sort(jobInfos, new Comparator<JobInfo>() {
            @Override
            public int compare(JobInfo o1, JobInfo o2) {
                return o1.getSn() - o2.getSn();
            }
        });
        boolean find = false;
        for (JobInfo job : jobInfos) {
            if (!find) {
                if (job.getRegionName().equals(jobInfo.getRegionName())
                        && job.getDeployName().equals(jobInfo.getDeployName())
                        && job.getGroupName().equals(jobInfo.getGroupName())
                        && job.getJobName().equals(jobInfo.getJobName())) {
                    find = true;
                }
                continue;
            }
            if (find) {
                String jobKey = DataKeys.getJobKey(job.getRegionName(), job.getDeployName(), job.getGroupName(),
                        job.getJobName());
                String isRun = masterNode.getDataItem(DataKeys.KEY_JOB_RUN, jobKey);
                if (isRun != null) {
                    masterNode.runJob(job.getRegionName(), job.getDeployName(), job.getGroupName(), job.getJobName(),
                            time, false);
                    return;
                }
            }
        }
    }

    /**
     * 定时器时间到了，触发执行
     *
     * @param time
     */
    @Override
    public void doWork(long time) {
        logger.info("doWork for job:" + time);
        List<JobInfo> jobInfos = jobConf.getGroupJobInfo(jobInfo.getRegionName(), jobInfo.getDeployName(),
                jobInfo.getGroupName());
        if (jobInfos == null || jobInfos.isEmpty()) {
            return;
        }
        Collections.sort(jobInfos, new Comparator<JobInfo>() {
            @Override
            public int compare(JobInfo o1, JobInfo o2) {
                return o1.getSn() - o2.getSn();
            }
        });
        // 执行组中第1个job
        for (JobInfo jobInfo : jobInfos) {
            String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                    jobInfo.getJobName());
            String isRun = masterNode.getDataItem(DataKeys.KEY_JOB_RUN, jobKey);
            if (isRun != null) {
                masterNode.runJob(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                        jobInfo.getJobName(), time, false);
                return;
            }
        }
    }

}
